package com.etl.service.impl;


import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.etl.constants.CommonConstant;
import com.etl.constants.UserSession;
import com.etl.entity.*;
import com.etl.enums.ResultEnum;
import com.etl.exception.GlobalException;
import com.etl.mapper.*;
import com.etl.model.rev.EtlTaskListRev;
import com.etl.service.IEtlTaskIncrementService;
import com.etl.service.KettleService;
import com.etl.util.AESUtil;
import com.etl.util.CodeGeneratedUtil;
import com.etl.util.DateUtil;
import com.etl.util.DbUntil;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.BeanUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import javax.websocket.Session;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by Administrator on 2019/10/21.
 */
@Service
@Slf4j
public class EtlTaskIncrementServiceImpl implements IEtlTaskIncrementService {


    @Resource
    private EtlTaskIncrementMapper etlTaskIncrementMapper;

    @Resource
    private EtlBusinessBatchMapper etlBusinessBatchMapper;

    @Resource
    private EtlBusinessDbInfoMapper dbInfoMapper;

    @Resource
    private EtlBusinessDbInfoMapper etlBusinessDbInfoMapper;

    @Resource
    private EtlTaskMainHistoryMapper etlTaskMainHistoryMapper;

    @Resource
    private EtlTaskBatchMapper etlTaskBatchMapper;

    @Resource
    private KettleService kettleService;



    /**
     * 获取所有增量任务
     * @param etlTaskListRev
     * @return
     */
    @Override
    public PageInfo<EtlTaskIncrement> getEtlIncrementTaskList(EtlTaskListRev etlTaskListRev) {
        PageHelper.startPage(etlTaskListRev.getPages(), etlTaskListRev.getSizes());
        QueryWrapper<EtlTaskIncrement> queryWrapper = new QueryWrapper<>();
        if (etlTaskListRev.getBatch() != null) {
            queryWrapper.eq("batch", etlTaskListRev.getBatch());
        }
        if (etlTaskListRev.getExecflag() != null) {
            queryWrapper.eq("exec_flag", etlTaskListRev.getExecflag());
        }
        if (Strings.isNotBlank(etlTaskListRev.getSrctable())) {
            queryWrapper.like("source_table_name", etlTaskListRev.getSrctable().trim());
        }

        //普通用户的话,只能查询自己的任务
        if(UserSession.getUser().getUserType() == 0){
            queryWrapper.eq("create_account",UserSession.getUser().getAccountName());
        }


        queryWrapper.orderByDesc("create_time");
        List<EtlTaskIncrement> etlMainTasks = etlTaskIncrementMapper.selectList(queryWrapper);
        PageInfo<EtlTaskIncrement> pageInfoDemo = new PageInfo<>(etlMainTasks);
        return pageInfoDemo;
    }

    @Override
    public void addEtlIncrementTask(EtlTaskIncrement etlTaskIncrement) {
//        String sourceTableName, String querySql, Integer batch, String comment,Integer tableType;
        /**
         * 查询批次号
         */
        EtlBusinessBatch etlBusinessBatch = etlBusinessBatchMapper.selectById(etlTaskIncrement.getBatch());
        if (etlBusinessBatch == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        //获取目标数据库
        EtlBusinessDbInfo etlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlBusinessBatch.getTargetDbId());
        if(etlBusinessDbInfo == null){
            throw new GlobalException(ResultEnum.DB_NOT_EXIST);
        }


        EtlBusinessDbInfo sourceEtlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlBusinessBatch.getSourceDbId());
        if (sourceEtlBusinessDbInfo == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        EtlBusinessDbInfo targetEtlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlBusinessBatch.getTargetDbId());
        if (targetEtlBusinessDbInfo == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }


        etlTaskIncrement.setBatchName(etlBusinessBatch.getBatchName());
        etlTaskIncrement.setSourceDbId(etlBusinessBatch.getSourceDbId());
        etlTaskIncrement.setSourceDbSysname(sourceEtlBusinessDbInfo.getDbSysname());
        etlTaskIncrement.setSourceTableName(etlTaskIncrement.getSourceTableName());


        etlTaskIncrement.setTargetDbId(etlBusinessBatch.getTargetDbId());
        etlTaskIncrement.setTargetDbSysname(etlBusinessBatch.getTargetDbSysname());
        etlTaskIncrement.setTargetTableName(etlTaskIncrement.getTargetTableName());


        etlTaskIncrement.setFlag(1);
        etlTaskIncrement.setCommitSize(CommonConstant.COMMITSIZE);   // 每多少行提交一次
        etlTaskIncrement.setExecFlag(CommonConstant.EXECFLAG);  // 任务执行状态 ，0任务运行成功等待下次执行，1任务运行中,2运行错误(问题修复后需手动修改为0程序才可再次执行)
        etlTaskIncrement.setSql1(etlTaskIncrement.getSql1());
        etlTaskIncrement.setSql1(etlTaskIncrement.getSql2());
        etlTaskIncrement.setSql1(etlTaskIncrement.getSql3());
        etlTaskIncrement.setTableComment(etlTaskIncrement.getTableComment());

        etlTaskIncrement.setCreateTime(new Date());
        //持久化
        int insert = etlTaskIncrementMapper.insert(etlTaskIncrement);

        log.info("新增任务:"+ etlTaskIncrement.getKid());
        //this.executeSingle(etlMainTask.getKid(),"创建表结构作业","创建表结构作业");

    }


    @Override
    public void updateEtlIncrementTask(EtlTaskIncrement etlTaskIncrement) {
        QueryWrapper<EtlTaskIncrement> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("kid", etlTaskIncrement.getKid());
        etlTaskIncrementMapper.update(etlTaskIncrement, queryWrapper);

        //创建表结构
//        this.executeSingle(etlMainTask.getKid(),"创建表结构作业","创建表结构作业");

    }



    @Transactional
    public void executeIncrementSingle(Long kid,String kettleHome,String kettleJobName) {

        /**
         * 检验任务
         */
        QueryWrapper<EtlTaskIncrement> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("KID", kid);
        EtlTaskIncrement etlTaskIncrement = etlTaskIncrementMapper.selectOne(queryWrapper);
        if (etlTaskIncrement == null) {
            log.error("【执行单个任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
//        if (etlUpdateTask.getExecFlag() == 1) {
//            log.error("【执行单个任务】-{}", ResultEnum.TASK_IS_RUNNING.getErrMsg());
//            throw new GlobalException(ResultEnum.TASK_IS_RUNNING);
//        }


        //首页获取上次同步时间
        String dataLastExtractTime = "2010-01-01 00:00:00";
        if(etlTaskIncrement.getDataLastExtractTime() != null){
            dataLastExtractTime = DateUtil.format(etlTaskIncrement.getDataLastExtractTime(),"yyyy-MM-dd HH:mm:ss");
        }


        /**
         * 获取业务系统数据库连接
         */
        //来源数据库信息
        EtlBusinessDbInfo sourceDbInfo = dbInfoMapper.selectById(etlTaskIncrement.getSourceDbId());
        //目标数据库信息
        EtlBusinessDbInfo tgtDbInfo = dbInfoMapper.selectById(etlTaskIncrement.getTargetDbId());

        Date localDate = new Date();
        //更新任务状态 execflag = 0 等待执行
        etlTaskIncrement.setExecFlag(0);
//        etlTaskIncrement.setDataLastExtractTime(localDate);
        etlTaskIncrementMapper.updateById(etlTaskIncrement);
        List<Long> kids = Lists.newArrayList();
        kids.add(kid);

        try {
            //设置变量值
            Map<String,String> jobVariableMap = new HashMap<>();
            jobVariableMap.put("kid",kid.toString());
            jobVariableMap.put("commit_size","1000");
            jobVariableMap.put("target_table_name",etlTaskIncrement.getTargetTableName());
            jobVariableMap.put("target_table_primary_id",etlTaskIncrement.getTargetTablePrimaryId());


            ///查询表数据字段
            jobVariableMap.put("sql1", String.format(etlTaskIncrement.getSql1(),dataLastExtractTime) );
            //根据时间查询源库要同步的数据数量
            jobVariableMap.put("sql2", String.format(etlTaskIncrement.getSql2(),dataLastExtractTime) );
            //查询需要新增的主键id
            jobVariableMap.put("sql3",String.format(etlTaskIncrement.getSql3(),dataLastExtractTime) );

            int error = kettleService.executeTask(jobVariableMap,kettleHome,kettleJobName,sourceDbInfo,tgtDbInfo, kids,"increment");
            if (error > 0) {
                throw new GlobalException(500, "作业执行出错");
            }else {
                etlTaskIncrement.setExecFlag(3);
                etlTaskIncrement.setDataLastExtractTime(localDate);
                etlTaskIncrementMapper.updateById(etlTaskIncrement);
            }


        } catch (Exception e) {
            log.error("ETL错误通知", "【" + etlTaskIncrement.getSourceDbSysname() + "】ETL抽取任务失败", "【" + etlTaskIncrement.getSourceDbSysname() + "】抽取任务失败", "【" + etlTaskIncrement.getTargetTableName() + "】表任务失败");
            e.printStackTrace();
            throw new GlobalException(500, e.getMessage());
        }


    }






    @Async
    @Override
    public void socketExecuteSingle(Session session, Long kid) {
        try {
            executeIncrementSingle(kid,"增量抽取作业","增量抽取作业");
            session.getBasicRemote().sendText("etlTaskDone");
        } catch (Exception e) {
            try {
                session.getBasicRemote().sendText("etlTaskErr");
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }

    }



    /**
     * 执行批次下的所有任务
     *
     * @param batch
     */
    @Override
    @Transactional
    public void executeAll(Integer batch, String accountName) {

        /**
         * 查询批次下所有任务
         */
        QueryWrapper<EtlTaskIncrement> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("batch", batch);
        List<EtlTaskIncrement> etlTaskIncrementList = etlTaskIncrementMapper.selectList(queryWrapper);
        if (etlTaskIncrementList == null || etlTaskIncrementList.size() == 0) {
            log.error("【执行批次下所有任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        //循环执行
        for (EtlTaskIncrement etlTaskIncrement : etlTaskIncrementList) {
            executeIncrementSingle(etlTaskIncrement.getKid(),"增量抽取作业","增量抽取作业");
        }



    }

    /**
     * 延时执行批次下的所有任务
     * 也就是将任务插入到定时任务表中，等待扫描器扫描并执行
     */
    @Override
    public void delayExecuteAll(Integer batch) {
        QueryWrapper<EtlBusinessBatch> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("batch", batch);
        EtlBusinessBatch etlBusinessBatch = etlBusinessBatchMapper.selectOne(queryWrapper);
        if (etlBusinessBatch == null) {
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        EtlTaskBatch etlTaskBatch = new EtlTaskBatch();
        etlTaskBatch.setBatch(batch);
        etlTaskBatch.setTaskId(CodeGeneratedUtil.genUUID());
        etlTaskBatch.setWaitRunTime(new Date());
        etlTaskBatch.setAccountName(UserSession.getUser().getAccountName());
        etlTaskBatch.setCreateTime(new Date());
        etlTaskBatchMapper.insert(etlTaskBatch);
    }





    /**
     * 删除任务和表结构
     *
     * @param kid
     */
    @Override
    @Transactional
    public void removeTaskAndTable(Long kid, Integer type) {

        String accountName = UserSession.getUser().getAccountName();
        EtlTaskIncrement etlTaskIncrement = etlTaskIncrementMapper.selectById(kid);
        if (etlTaskIncrement == null) {
            log.error("【删除任务】-{}", ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST.getErrMsg());
            throw new GlobalException(ResultEnum.BATCH_SYSTEM_NAME_NOT_EXIST);
        }
        /**
         * 删除表
         */

        try {
            String tableName1 = etlTaskIncrement.getTargetTableName().trim();
            String delDate = DateUtil.getCurrent("yyMMddHHmmss");

            EtlBusinessDbInfo etlBusinessDbInfo = etlBusinessDbInfoMapper.selectById(etlTaskIncrement.getTargetDbId());
            //解密数据库密码
            etlBusinessDbInfo.setPassword(AESUtil.decrypt(etlBusinessDbInfo.getPassword(), AESUtil.password));

            //先判断是否存在该表，存在则删除
            String tableSql1 = "";
            String renameTable1 = "ALTER TABLE "+tableName1+" RENAME TO "+tableName1 + "_"+ accountName+"_DEL_" + delDate;

            if(CommonConstant.ORACLE_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //oracle

                tableSql1 = "select table_name tableName from user_tables where table_name = '"+tableName1.toUpperCase()+"'";

            }else if(CommonConstant.MYSQL_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //mysql
                tableSql1 = "select table_name tableName from information_schema.TABLES  where table_name = '"+tableName1+"'" +"and table_schema='"+etlBusinessDbInfo.getDbSchema()+"'";

            }else if(CommonConstant.MSSQL_NAME.equalsIgnoreCase(etlBusinessDbInfo.getDbType())){ //mssql

                tableSql1 = "SELECT name tableName FROM sysobjects where name='"+tableName1+"'";
                renameTable1 = "exec sp_rename "+tableName1+","+tableName1 + "_"+ accountName+"_DEL_" + delDate;
            }
            //判断表是否存在,存在则重命名删除
            String tabelName1 = DbUntil.getTabelName(tableSql1, etlBusinessDbInfo);
            if(StringUtils.isNotBlank(tabelName1)){
                DbUntil.executeSql(renameTable1,etlBusinessDbInfo);
            }



        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }

        /**
         * 删除任务
         */
        if (type != null && type == 1) {
            etlTaskIncrementMapper.deleteById(kid);
            /**
             * 将删除的任务插入到历史表里
             */
            EtlTaskMainHistory etlTaskMainHistory = new EtlTaskMainHistory();
            BeanUtils.copyProperties(etlTaskIncrement, etlTaskMainHistory);
            etlTaskMainHistory.setKid(kid);
            etlTaskMainHistoryMapper.insert(etlTaskMainHistory);
        }


    }



}

